package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author sw
 * @CreateTime 2025-02-28
 */
public class CommentApp extends BaseSqlApp {
    public static void main(String[] args) {
        new CommentApp().start(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO,4,10012);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        //1、读取ods层数据
        readOdsDb(tableEnv,groupId);
        //2、过滤评论数据
        Table comment = getComment(tableEnv);
        tableEnv.createTemporaryView("comment_info",comment);
//        comment.execute().print();
        //3、字典数据,基类
        //4、读取码表
        createBaseDic(tableEnv);
        //5、关联数据
        Table table = getTable(tableEnv);
        table.execute().print();
//        //6、写入到kafka
        extracted(tableEnv);
//        //写出去
        table.insertInto(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO).execute();
    }

    private static void extracted(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("create table " + Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO + "(\n" +
                "id string,\n" +
                "user_id string,\n" +
                "nick_name string,\n" +
                "sku_id string,\n" +
                "spu_id string,\n" +
                "order_id string,\n" +
                "appraise_code string,\n" +
                "appraise_name string,\n" +
                "comment_txt string,\n" +
                "create_time string\n" +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_INTERACTION_COMMENT_INFO));
    }

    /**
     * 5、关联数据
     * @param tableEnv
     * @return
     */
    private static Table getTable(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("SELECT \n" +
                " id,\n" +
                " user_id,\n" +
                " nick_name,\n" +
                " sku_id,\n" +
                " spu_id,\n" +
                " order_id,\n" +
                " appraise,\n" +
                " info.dic_name,\n" +
                " comment_txt,\n" +
                " create_time \n" +
                " FROM comment_info AS c \n" +
                " JOIN base_dic FOR SYSTEM_TIME AS OF c.proc_time AS b \n" +
                " ON c.appraise = b.rowkey");
    }

    /**
     * 2、过滤评论数据
     * @param tableEnv
     * @return
     */
    private static Table getComment(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['nick_name'] nick_name,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['spu_id'] spu_id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['appraise'] appraise,\n" +
                "`data`['comment_txt'] comment_txt,\n" +
                "`data`['create_time'] create_time,\n" +
                "proc_time\n" +
                "from topic_db\n" +
                "where `database` = 'gmall_realtime' and `type` = 'bootstrap-insert'\n" +
                "and `table` = 'comment_info'");
    }
}
